require('dotenv').config();

const moment = require('moment');
const WebSocket = require('ws');
const pako = require('pako');
const SocksProxyAgent = require('socks-proxy-agent');
const redis = require('./ioredis');

const WS_URL = 'wss://api.huobi.pro/ws';
const symbols = require('../config/symbols');
const periods = require('../config/periods');

let ws;
let connecting = false;
let timestamp = 0;
let timer = [];

// 创建连接
function connect() {
    if (process.env.APP_ENV === 'production') {
        ws = new WebSocket(WS_URL);
    } else {
        const agent = new SocksProxyAgent(process.env.PROXY);
        ws = new WebSocket(WS_URL, { agent: agent });
    }
    registerEventListener();
}

// 重新连接
function reconnect() {
    timer.forEach(item => clearTimeout(item));
    if (connecting) return;
    connecting = true;
    setTimeout(() => {
        connect();
        connecting = false;
    }, 1000);
}

function registerEventListener() {
    ws.on('open', () => {
        console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] open`);
        initKline();
        subscribe(ws);
    });

    ws.on('message', data => {
        let text = pako.inflate(data, { to: 'string' });
        let msg = JSON.parse(text);
        if (msg.ping) {
            timestamp = msg.ping;
            ws.send(JSON.stringify({ pong: timestamp }));
        } else if (msg.tick) {
            subscribeHandle(msg);
        } else if (msg.rep) {
            requestHandle(msg);
        } else {
            console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] ${text}`);
        }
    });

    ws.on('close', () => {
        console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] close`);
        reconnect();
    });

    ws.on('error', error => {
        console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] ${error}`);
        reconnect();
    });
}

// K线数据初始化
function initKline() {
    for (let i = 0; i < symbols.length; i++) {
        const symbol = symbols[i];
        for (let j = 0; j < periods.length; j++) {
            const period = periods[j];
            let t = setTimeout(() => {
                ws.send(JSON.stringify({
                    req: `market.${symbol}.kline.${period}`,
                    id: `${symbol}`
                }));
            }, (i + j + 1) * 1000);
            timer.push(t);
        }
    }
}

// 订阅
function subscribe(ws) {
    for (let symbol of symbols) {
        // 订阅深度
        // 谨慎选择合并的深度，ws每次推送全量的深度数据，若未能及时处理容易引起消息堆积并且引发行情延时
        ws.send(JSON.stringify({
            "sub": `market.${symbol}.depth.step0`,
            "id": `${symbol}`
        }));

        // 订阅K线
        for (let i = 0; i < periods.length; i++) {
            const period = periods[i];
            ws.send(JSON.stringify({
                "sub": `market.${symbol}.kline.${period}`,
                "id": `${symbol}`
            }));
        }

        // 此主题提供市场最新成交明细。
        ws.send(JSON.stringify({
            "sub": `market.${symbol}.trade.detail`,
            "id": `${symbol}`
        }));

        // 此主题提供24小时内最新市场概要。
        ws.send(JSON.stringify({
            "sub": `market.${symbol}.detail`,
            "id": `${symbol}`
        }));
    }
}

// 订阅消息处理器
function subscribeHandle(data) {
    const symbol = data.ch.split('.')[1];
    const channel = data.ch.split('.')[2];
    switch (channel) {
        case 'depth':
            console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] depth:${symbol}`);
            depthHandle(symbol, data);
            break;
        case 'kline':
            const period = data.ch.split('.')[3];
            console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] kline:${symbol}:${period}`);
            klineHandle(symbol, period, data);
            break;
        case 'trade':
            console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] trade:${symbol}`);
            tradeHandle(symbol, data);
            break;
        case 'detail':
            console.log(`[${moment().format('YYYY-MM-DD HH:mm:ss')}] ticker:${symbol}`);
            tickerHandle(symbol, data);
            break;
    }
}

// 请求消息处理器
function requestHandle(data) {
    const symbol = data.rep.split('.')[1];
    const channel = data.rep.split('.')[2];
    switch (channel) {
        case 'kline':
            const period = data.rep.split('.')[3];
            klineHandle(symbol, period, data);
            break;
        default:
            break;
    }
}

// 深度数据处理器
function depthHandle(symbol, data) {
    const asks = data.tick.asks.slice(4, 104);
    asks.map(item => item.reverse());
    asks.unshift(`huobi:depth:${symbol}:asks`);
    redis.del(`huobi:depth:${symbol}:asks`);
    redis.zadd.apply(redis, asks);

    const bids = data.tick.bids.slice(4, 104);
    bids.map(item => item.reverse());
    bids.unshift(`huobi:depth:${symbol}:bids`);
    redis.del(`huobi:depth:${symbol}:bids`);
    redis.zadd.apply(redis, bids);
}

// 交易数据处理器
function tradeHandle(symbol, data) {
    for (let trade of data.tick.data) {
        redis.lpush(`huobi:trade:${symbol}`, JSON.stringify({
            id: data.tick.id,
            side: trade.direction.toUpperCase(),
            price: trade.price.toFixed(8),
            number: trade.amount.toFixed(8),
            created_at: moment(data.tick.ts).format('YYYY-MM-DD HH:mm:ss'),
        }));
    }
    redis.ltrim(`huobi:trade:${symbol}`, 0, 99);
}

// 行情数据处理器
function tickerHandle(symbol, data) {
    redis.set(`huobi:ticker:${symbol}`, JSON.stringify({
        open: data.tick.open.toFixed(8),
        close: data.tick.close.toFixed(8),
        high: data.tick.high.toFixed(8),
        low: data.tick.low.toFixed(8),
        number: data.tick.amount.toFixed(8),
        total: data.tick.vol.toFixed(8),
    }));
}

// K线数据处理器
function klineHandle(symbol, period, data) {
    if (data.tick) {
        // 订阅数据
        redis.kline(
            `huobi:kline:${symbol}:${period}`,
            data.tick.id,
            JSON.stringify([
                data.tick.id,
                data.tick.open.toFixed(8),
                data.tick.high.toFixed(8),
                data.tick.low.toFixed(8),
                data.tick.close.toFixed(8),
                data.tick.amount.toFixed(8),
                data.tick.vol.toFixed(8),
                data.tick.count,
            ])
        );
    } else if (data.data) {
        // 请求数据
        data.data.forEach(item => {
            redis.kline(
                `huobi:kline:${symbol}:${period}`,
                item.id,
                JSON.stringify([
                    item.id,
                    item.open.toFixed(8),
                    item.high.toFixed(8),
                    item.low.toFixed(8),
                    item.close.toFixed(8),
                    item.amount.toFixed(8),
                    item.vol.toFixed(8),
                    item.count,
                ])
            );
        });
    }
}

connect();

// 定时检查连接状态
setInterval(() => {
    // 连接已中断，立即重新连接
    if (ws.readyState === WebSocket.CLOSED) {
        reconnect();
        return;
    }
    // 连接未中断，但超过 10 秒未收到消息，关闭连接
    if (Date.now() - timestamp > 10000) {
        ws.close();
    }
}, 10 * 1000); // 每10秒执行

// K线数据裁剪
setInterval(() => {
    symbols.forEach(symbol => {
        periods.forEach(period => {
            redis.zremrangebyrank(`huobi:kline:${symbol}:${period}`, 0, -1001);
        });
    });
}, 60 * 60 * 1000); // 每小时执行
